package com.atguigu.userprofile.app

import java.util
import java.util.Properties

import com.alibaba.fastjson.JSONObject
import com.atguigu.userprofile.bean.{TagInfo, TaskInfo, TaskTagRule}
import com.atguigu.userprofile.dao.{TagDAO, TaskDAO, TaskTagRuleDAO}
import com.atguigu.userprofile.util.{MyPropertiesUtil, MySqlUtil}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

object TaskSQLApp {

  def main(args: Array[String]): Unit = {
     val sparkConf: SparkConf = new SparkConf().setAppName("task_sql_app")
       //.setMaster("local[*]")
     val sparkSession: SparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()

     val properties: Properties = MyPropertiesUtil.load("config.properties")
     val hdfsStorePath: String = properties.getProperty("hdfs-store.path")
     val dwDBName: String = properties.getProperty("data-warehouse.dbname")
     val upDBName: String = properties.getProperty("user-profile.dbname")

     println(dwDBName)
     println(upDBName)

//    val list: util.List[JSONObject] = MySqlUtil.queryList("select * from task_info")
//    println(list )


    //TODO : 可以通过main的参数 获得2个业务参数 一个是task_id , 一个 task_date
    var taskId=""
    var taskDate=""
    if(args.length>0){
      taskId=args(0)
      taskDate=args(1)
    }
    //TODO： 通过task_id ，标签信息、任务信息、规则信息
    //查询mysql
    //查询标签信息sql
    //任务信息的sql
    // 查询规则信息的sql
     val tagInfo : TagInfo  = TagDAO.getTagInfo(taskId);

    val taskInfo: TaskInfo = TaskDAO.getTaskInfo(taskId)

      val tagRuleList: List[TaskTagRule] = TaskTagRuleDAO.getTaskTagRuleListByTaskId(taskId)

     println(tagInfo)
     println(taskInfo)

    //组织建表语句   依据：1 该标签的tag_code as 表名
                     //  2 该标签的tag_value_type 作为表中 tag_value字段的类型 （String ,decimal ,bigint ,String)
                     //   3 分区 用taskDate 日期进行分区
                     //  4   存储格式  文本 \t 分割
                    //  5 存储的位置  hdfsStorePath+userprofileDbName+tableName
                    //   6  备注信息 tag中文名

    val tableName=tagInfo.tagCode.toLowerCase()
    val tagName= tagInfo.tagName
    val storePath=hdfsStorePath+"/"
    var tagValueType=  ""
    //1 整数 2 浮点 3 文本 4 日期
    if(tagInfo.tagValueType =="1"){
      tagValueType=   "bigint"
    }else if (tagInfo.tagValueType =="2"){
      tagValueType=   "decimal"
    }  else if (tagInfo.tagValueType =="3"){
      tagValueType=   "string"
    }   else if(tagInfo.tagValueType =="4"){
      tagValueType=   "string"
    }


     val createTableSQL=
       s"""
          |create  table if not exists $tableName
          |(   uid string ,
          |    tag_value  $tagValueType
          |) comment '$tagName'
          |  ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t'
          |  PARTITIONED BY (  `dt`  string )
          |   location '$hdfsStorePath/$upDBName/$tableName'
        """.stripMargin

    println(createTableSQL)
    //
    //select   id as  uid,
    // if(gender is null ,'U',gender) as  query_value
    // from   dim_user_info   where  dt='9999-99-99'
    // 动态拼接select 部分

    //根据规则 组合 caseWhenSQL
    //有规则的  非连续值
//     	caseWhenSql=
//      when  'M'  then  '男’
//    when 'F' then '女'
//    when 'U' then '未知'   ===>  when

    var taskSQL=taskInfo.taskSql
    //替换掉 $dt
    taskSQL=taskSQL.replace("$dt",taskDate)

    var tagValueSql=""
    if(tagRuleList.size>0){

      val caseWhenList: List[String] = tagRuleList.map { taskTagRule =>
        s"when '${taskTagRule.queryValue}' then '${taskTagRule.subTagValue}'"
      }
       val  caseWhenSQL  = caseWhenList.mkString(" ")
      tagValueSql=s"case  query_value $caseWhenSQL end as tag_value"

    }else{ //无规则的   连续值  金额 个数  时间
      tagValueSql= "query_value as tag_value"
    }

    //  select
    //   uid,
    //    case  query_value $caseWhenSql end as tag_value
    //   from (
    //     select   id as  uid,
    //     if(gender is null ,'U',gender) as  query_value
    //     from   dim_user_info   where  dt='9999-99-99'
    //	)


    var selectSql=
      s"""
         |  select
         |   uid,
         |    $tagValueSql
         |   from (
         |       $taskSQL
         |	)
       """.stripMargin

    println(selectSql)

    // insert+ select 拼接
    //insert overwrite table $tag_code partition by (dt='$taskDate')
    val insertSql=s"insert overwrite table $upDBName.$tableName partition  (dt='$taskDate') $selectSql"

    println(insertSql)

    sparkSession.sql("use "+upDBName)
    sparkSession.sql(createTableSQL)

    sparkSession.sql("use "+dwDBName)
    sparkSession.sql(insertSql)




  }

}
